package dk.almbrand.microservice.service.base;

import dk.almbrand.microservice.cache.Cache;
import dk.almbrand.microservice.circuitbreaker.exceptions.ArgumentOutOfRangeException;
import dk.almbrand.microservice.exception.SystemException;
import dk.almbrand.microservice.heartbeat.Heartbeat;
import dk.almbrand.microservice.heartbeat.HeartbeatImpl;
import dk.almbrand.microservice.kafka.KafkaFacade;
import dk.almbrand.microservice.kafka.KafkaFacadeImpl;
import dk.almbrand.microservice.model.Event;
import dk.almbrand.microservice.model.RequestEvent;
import dk.almbrand.microservice.model.ResponseEvent;
import dk.almbrand.microservice.util.GsonUtils;

public abstract class MicroserviceWrapper {
    
    public static final String INVALID_MESSAGES_TOPIC = "invalid_messages";
    
    private Heartbeat heatbeat;
    private KafkaFacade kafka;
    
    /**
     *
     * @param host
     * @param serializer
     * @param topic
     * @throws ArgumentOutOfRangeException
     */
    public MicroserviceWrapper(String host, String serializer, String topic) throws ArgumentOutOfRangeException, SystemException {
        kafka = new KafkaFacadeImpl(this, host, serializer, topic);
        heatbeat = new HeartbeatImpl(kafka).build();
    }

    public final void publish(Event event, String topic) throws SystemException {
        String data = "";

        if(topic == null || topic == "") {
            throw new SystemException("No topic has been defined"); 
        }
        
        if(event != null) {
            if (event instanceof ResponseEvent) {
                data = new GsonUtils<ResponseEvent>().toJSON((ResponseEvent)event);
                Cache.getInstance().getResponseEventCache().add(event);
            } else if (event instanceof RequestEvent) {
                data = new GsonUtils<RequestEvent>().toJSON((RequestEvent)event);
                Cache.getInstance().getRequestEventCache().add(event);
            }
            
            if(data != null && !data.equals("")) {
                getKafka().getProducer().publish(topic, event.getKey(), data);
            } else {
                throw new SystemException("data object was null or empty");
            }
        }
    }

    // Skal deles ud i privat methode og en som bliver overridden af den nedarvende service.
    public final void onMessage(Event requestEvent) {
        if (requestEvent.getObject() instanceof String && String.valueOf(requestEvent.getObject()).equals(HeartbeatImpl.PING)) {
            heatbeat.setIsAlive(true);
            return;
        }

        // Messge has already been processed. Only caching, in case service is re-booted we start over.
        if (Cache.getInstance().getRequestEventCache().contains(requestEvent)) {
            return;
        }

        // Aimed at the end implementation.
        handle(requestEvent);
    }

    /**
     * Do nothing!! - Method should be overridden by the final microservice implementation, to catch and handle incomming messages.
     * @param Event
     */
    public abstract void handle(Event requestEvent);

    protected void logInvalidMessage(Event event, String message, String topic, String applicationId) {
        try {
            
            if(topic == null || topic.equals("")) {
                topic = INVALID_MESSAGES_TOPIC;
            }
            
            ResponseEvent response = buildResponseEvent(-1, message, event.getKey(), event.getJson(), applicationId);
            publish(response, topic);
        } catch (SystemException f) {
        }        
    }
    
    protected ResponseEvent buildResponseEvent(int resultCode, String description, String key, Object object, String applicationId) {
        ResponseEvent resp = new ResponseEvent();
        resp.setKey(key);
        resp.setResultCode(resultCode);
        resp.setDesc(description);
        resp.setApplicationId(applicationId);
                                                  
        if(object != null) {
            resp.setObject(object);
        }
        
        return resp;
    }
    
    private KafkaFacade getKafka() {
        return kafka;
    }


}
